/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by WangYunlai on 2023/06/28.
//

#include <sstream>
#include <vector>
#include <algorithm>
#include "sql/parser/value.h"
#include "storage/field/field.h"
#include "common/log/log.h"
#include "common/lang/string.h"
#include "storage/common/condition_filter.h"
#include "common/lang/comparator.h"

const char *ATTR_TYPE_NAME[] =
    {"undefined", "chars", "ints", "floats", "dates", "texts", "booleans"};

const char *attr_type_to_string(AttrType type) {
  // 改成小于BOOLEANS，这样每次添加新的类型时就只用改上面的ATTR_TYPE_NAME就行了
  if (type >= UNDEFINED && type < BOOLEANS) {
    return ATTR_TYPE_NAME[type];
  }
  return "unknown";
}
AttrType attr_type_from_string(const char *s) {
  for (unsigned int i = 0; i < sizeof(ATTR_TYPE_NAME) / sizeof(ATTR_TYPE_NAME[0]); i++) {
    if (0 == strcmp(ATTR_TYPE_NAME[i], s)) {
      return (AttrType)i;
    }
  }
  return UNDEFINED;
}

Value::Value(int val) {
  set_int(val);
}

Value::Value(float val) {
  set_float(val);
}

Value::Value(bool val) {
  set_boolean(val);
}

Value::Value(const char *s, int len /*= 0*/) {
  set_string(s, len);
}

Value::Value(int val, AttrType type) {
  switch (type) {
    case DATES: {
      set_date(val);
    } break;
    default: {
      set_int(val);
    } break;
  }
}

void Value::set_data(char *data, int length) {
  switch (attr_type_) {
    case TEXTS: {
      set_text_string(data, length);
    } break;
    case CHARS: {
      set_string(data, length);
    } break;
    case INTS: {
      num_value_.int_value_ = *reinterpret_cast<int *>(data);
      length_ = length;
    } break;
    case FLOATS: {
      num_value_.float_value_ = *reinterpret_cast<float *>(data);
      length_ = length;
    } break;
    case BOOLEANS: {
      num_value_.bool_value_ = *reinterpret_cast<int *>(data) != 0;
      length_ = length;
    } break;
    case DATES: {
      num_value_.date_value_ = *reinterpret_cast<int *>(data);
      length_ = length;
    } break;
    default: {
      LOG_WARN("unknown data type: %d", attr_type_);
    } break;
  }
}
void Value::set_int(int val) {
  attr_type_ = INTS;
  num_value_.int_value_ = val;
  length_ = sizeof(val);
}

void Value::set_null() {
  attr_type_ = NULLS;
}

void Value::set_float(float val) {
  attr_type_ = FLOATS;
  num_value_.float_value_ = val;
  length_ = sizeof(val);
}
void Value::set_boolean(bool val) {
  attr_type_ = BOOLEANS;
  num_value_.bool_value_ = val;
  length_ = sizeof(val);
}
void Value::set_string(const char *s, int len /*= 0*/) {
  attr_type_ = CHARS;
  if (len > 0) {
    len = strnlen(s, len);
    str_value_.assign(s, len);
  } else {
    str_value_.assign(s);
  }
  length_ = str_value_.length();
}

void Value::set_text_string(const char *s, int len /*= 0*/) {
  attr_type_ = TEXTS;
  if (len > 0) {
    len = strnlen(s, len);
    str_value_.assign(s, len);
  } else {
    str_value_.assign(s);
  }
  length_ = str_value_.length();
}

bool checkDate(int y, int m, int d) {
  static int mon[] = {0, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31};
  bool leap = (y%400 == 0 || (y%100 && y%4 == 0));
  return y > 0
         && (m > 0)&&(m <= 12)
         && (d > 0)&&(d <= ((m == 2 && leap)?1:0) + mon[m]);
}

int value_init_date(const char *val, int &ret) {
  int year, month, day;
  sscanf(val, "%d-%d-%d", &year, &month, &day);
  bool isValid = checkDate(year, month, day);
  if (!isValid) {
    return 1;
  }

  int time = (year*common::YEAR_WEIGHT + month*common::MONTH_WEIGHT
      + day*common::DAY_WEIGHT);
  ret = time;
  return 0;
}

void Value::set_date(int val) {
  attr_type_ = DATES;
  num_value_.date_value_ = val;
  length_ = sizeof(val);
}

void Value::set_page_num(PageNum val) {
  attr_type_ = TEXTS;
  num_value_.page_num_value_ = val;
  length_ = sizeof(PageNum);
}

void Value::set_value(const Value &value) {
  switch (value.attr_type_) {
    case INTS: {
      set_int(value.get_int());
    } break;
    case FLOATS: {
      set_float(value.get_float());
    } break;
    case CHARS: {
      set_string(value.get_string().c_str());
    } break;
    case BOOLEANS: {
      set_boolean(value.get_boolean());
    } break;
    case DATES: {
      set_date(value.get_date());
    } break;
    case TEXTS: {
      // 是在filter中调用
      set_text_string(value.get_string().c_str());
    } break;
    case UNDEFINED: {
      ASSERT(false, "got an invalid value type");
    } break;
  }
}

const char *Value::data() const {
  switch (attr_type_) {
    case CHARS: {
      return str_value_.c_str();
    } break;
    default: {
      return (const char *)&num_value_;
    } break;
  }
}

std::string Value::to_string() const {
  std::stringstream os;
  switch (attr_type_) {
    case INTS: {
      os << num_value_.int_value_;
    } break;
    case FLOATS: {
      os << common::double_to_str(num_value_.float_value_);
    } break;
    case BOOLEANS: {
      os << num_value_.bool_value_;
    } break;
//    case TEXTS: {
//      os << num_value_.page_num_value_;
//    } break;
    case TEXTS:
    case CHARS: {
      os << str_value_;
    } break;
    case DATES: {
      os << common::date_to_str(num_value_.date_value_);
    } break;
    case NULLS: {
      os << "NULL";
    } break;
    default: {
      LOG_WARN("unsupported attr type: %d", attr_type_);
    } break;
  }
  return os.str();
}

RC Value::compare_int(const Value &master, const Value &other,
                      int &result) const {
  switch (other.attr_type()) {
    case INTS: {
      result = common::compare_int((void *)&master.num_value_.int_value_,
         (void *)&other.num_value_.int_value_);
      return RC::SUCCESS;
    }
    case FLOATS: {
      float this_data = this->num_value_.int_value_;
      result = common::compare_float((void *)&this_data,
          (void *)&other.num_value_.float_value_);
      return RC::SUCCESS;
    }
    case CHARS: {
      std::istringstream iss(other.get_string());
//      int num = 0;
      float num = 0;
      iss >> num;
      float tmp = static_cast<float>(master.num_value_.int_value_);
      result = common::compare_float(&tmp, &num);
      return RC::SUCCESS;
    }
    default: {
      LOG_ERROR("未实现的比较 %s与%s", attr_type_to_string(master.attr_type_),
          attr_type_to_string(other.attr_type_));
      return RC::COMPARE_UNIMPLEMENT;
    }
  }
}

RC Value::compare_float(const Value &master,
                        const Value &other, int &result) const {
  switch (other.attr_type()) {
    case INTS: {
      float other_data = other.num_value_.int_value_;
      result = common::compare_float((void *)&this->num_value_.float_value_,
          (void *)&other_data);
      return RC::SUCCESS;
    }
    case FLOATS: {
      result = common::compare_float((void *)&this->num_value_.float_value_,
          (void *)&other.num_value_.float_value_);
      return RC::SUCCESS;
    }
    case CHARS: {
      std::istringstream iss(other.get_string());
      float num = 0;
      iss >> num;
      result = common::compare_int((void *)&master.num_value_.int_value_, &num);
      return RC::SUCCESS;
    }
    default: {
      LOG_ERROR("未实现的比较 %s与%s", attr_type_to_string(master.attr_type_),
          attr_type_to_string(other.attr_type_));
      return RC::COMPARE_UNIMPLEMENT;
    }
  }
}

RC Value::compare_char(const Value &master,
                       const Value &other, int &result) const {
  switch (other.attr_type()) {
    case INTS:
    case FLOATS: {
      std::istringstream iss(master.to_string());
      float num1 = 0;
      iss >> num1;
      float num2 = other.attr_type_== FLOATS ? other.num_value_.float_value_:(float)other.num_value_.int_value_;
      result = common::compare_float(&num1, &num2);
      return RC::SUCCESS;
//      const char *str;
//      str = std::to_string(other.get_int()).c_str();
//      return common::compare_string((void *)str, strlen(str),
//          (void *)other.str_value_.c_str(), other.str_value_.length());
    }
    case CHARS: {
      result = common::compare_string((void *)master.str_value_.c_str(),
          master.str_value_.length(),
          (void *)other.str_value_.c_str(),
          other.str_value_.length());
      return RC::SUCCESS;
    }
    default: {
      LOG_ERROR("未实现的比较 %s与%s", attr_type_to_string(master.attr_type_),
          attr_type_to_string(other.attr_type_));
      return RC::COMPARE_UNIMPLEMENT;
    }
  }
}

RC Value::compare(const Value &other, int &result) const
{
  RC rc = RC::SUCCESS;

  if (OB_SUCC(field_type_compare_compatible_table(this->attr_type_, other.attr_type_))) {
    switch (this->attr_type_) {
      case INTS: {
        rc = compare_int(*this, other, result);
      } break;
      case FLOATS: {
        rc = compare_float(*this, other, result);
      } break;
      case CHARS: {
        rc = compare_char(*this, other, result);
      } break;
      case BOOLEANS: {
        result = common::compare_int((void *)&this->num_value_.bool_value_,
            (void *)&other.num_value_.bool_value_);
      } break;
      case DATES: {
        result = common::compare_int((void *)&this->num_value_.int_value_,
            (void *)&other.num_value_.int_value_);
      } break;
      case TEXTS: {
        result = common::compare_string((void *)this->str_value_.c_str(),
            this->str_value_.length(),
            (void *)other.str_value_.c_str(),
            other.str_value_.length());
      } break;
      default: {
        LOG_WARN("unsupported type: %d", this->attr_type_);
        rc = RC::INVALID_ARGUMENT;
      } break;
    }
    return rc;
  }
  LOG_WARN("not supported");
  return RC::COMPARE_NOT_SUPPORT;
}

// LIKE算法开始

/**
 * 字符串匹配算法
 * 若成功匹配，则返回主串匹配成功处最后一个字符的下一个位置索引
 * 若匹配失败，则返回-1
*/
int KMP(std::string p, std::string s) {
  int n = s.size(), m = p.size();
  if (m == 0) return 0;
  s.insert(s.begin(), ' ');
  p.insert(p.begin(), ' ');
  std::vector<int> next(m + 1);
  for (int i = 2, j = 0; i <= m; i++) {
    while (j && p[i] != p[j + 1]) j = next[j];
      if (p[i] == p[j + 1]) j++;
      next[i] = j;
  }
  for (int i = 1, j = 0; i <= n; i++) {
    while (j && (s[i] != p[j + 1] && p[j + 1] != '_')) j = next[j];
      if (s[i] == p[j + 1] || p[j + 1] == '_') j++;
      if (j == m) return i;
  }
  return -1;
}

bool is_like(std::string s1, std::string s2) {
  std::stringstream stream_data(s2);
  std::vector<std::string> subs2;
  std::string res;
  // 将字符按%分割，相当于java的str.split("%")
  while (std::getline(stream_data, res, '%')) {
    subs2.push_back(res);
  }
  for (auto it = subs2.begin(); it != subs2.end();) {
    if (*it == "") {
      it = subs2.erase(it);
    } else {
      it++;
    }
  }
  int start = 0;
  for (int i = 0; i < subs2.size(); i++) {
    std::string s3 = s1.substr(start);
    int res = KMP(subs2[i], s3);
    if (res == -1) {
      return false;
    } else if (res != -1) {  // 若其开头不是%，则必须从主串开头即匹配成功，若其结尾不是%也同理
      if (i == 0 && res != subs2[0].size() && s2[0] != '%') return false;
      if (i == subs2.size() - 1 && res != s3.size() && s2[s2.size() - 1] != '%')
        return false;
    }
    start += res;
  }
  return true;
}

bool Value::compare_like(const Value &other) const {
  // no need to check if the type is valid. function compare_value in expression.cpp guarantees.
  std::string s1 = this->str_value_;
  std::string s2 = other.str_value_;
  if (s1.size() == 0 && s2.size() == 0) return true;
  if (s1.size() != 0 && s2.size() == 0) return false;

  int cnt_ = 0;
  int cnto = 0;
  for (int i = 0; i < s2.size(); i++) {
    if (s2[i] == '_') cnt_++;
    else if (s2[i] == '%') cnto++;
  }
  // if all chars in s2 is '_' or '%'
  if (cnt_ + cnto == s2.size()) {
    if (cnto == s2.size()) {
      return true;
    } else if (cnt_ == s2.size()) {
      if (cnt_ == s1.size()) return true;
      else
        return false;
    } else {
        if (cnt_ > s1.size()) return false;
        else
          return true;
      }
  }
  // 翻转后再次比较，处理形如abbc与%b_这种的匹配
  std::string s3(s1);
  std::string s4(s2);
  std::reverse(s3.begin(), s3.end());
  std::reverse(s4.begin(), s4.end());
  return is_like(s1, s2) || is_like(s3, s4);
}
// LIKE算法结束

RC Value::compare_in(const std::vector<Value> &other, bool &result) const {
  int res = 0;
  RC rc = RC::SUCCESS;
  for (int i = 0; i < other.size(); i++) {
    if (other[i].attr_type() == NULLS) {
      continue;
    }
    if (this->attr_type() == NULLS) {
      result = false;
      return RC::SUCCESS;
    }
    rc = compare(other[i], res);
    if (rc != RC::SUCCESS) {
      return rc;
    }
    if (res == 0) {
      result = true;
      break;
    }
  }
  return RC::SUCCESS;
}


int Value::get_int() const {
  switch (attr_type_) {
    case CHARS: {
      try {
        return (int)(std::stol(str_value_));
      } catch (std::exception const &ex) {
        LOG_TRACE("failed to convert string to number. s=%s, ex=%s",
            str_value_.c_str(), ex.what());
        return 0;
      }
    }
    case INTS: {
      return num_value_.int_value_;
    }
    case FLOATS: {
      return (int)(num_value_.float_value_);
    }
    case BOOLEANS: {
      return (int)(num_value_.bool_value_);
    }
    case DATES: {
      return num_value_.date_value_;
      return 0;
    }
    default: {
      LOG_WARN("unknown data type. type=%d", attr_type_);
      return 0;
    }
  }
  return 0;
}

float Value::get_float() const {
  switch (attr_type_) {
    case CHARS: {
      try {
        return std::stof(str_value_);
      } catch (std::exception const &ex) {
        LOG_TRACE("failed to convert string to float. s=%s, ex=%s", str_value_.c_str(), ex.what());
        return 0.0;
      }
    }
    case INTS: {
      return float(num_value_.int_value_);
    }
    case FLOATS: {
      return num_value_.float_value_;
    }
    case BOOLEANS: {
      return float(num_value_.bool_value_);
    }
    case DATES: {
      LOG_ERROR("can't convert date to float");
      return 0;
    }
    default: {
      LOG_WARN("unknown data type. type=%d", attr_type_);
      return 0;
    }
  }
  return 0;
}

std::string Value::get_string() const {
  return this->to_string();
}

bool Value::get_null() const {
  return attr_type_ == NULLS;
}
bool Value::get_boolean() const {
  switch (attr_type_) {
    case CHARS: {
      try {
        float val = std::stof(str_value_);
        if (val >= EPSILON || val <= -EPSILON) {
          return true;
        }

        int int_val = std::stol(str_value_);
        if (int_val != 0) {
          return true;
        }

        return !str_value_.empty();
      } catch (std::exception const &ex) {
        LOG_TRACE("failed to convert string to float or integer. s=%s, ex=%s",
            str_value_.c_str(), ex.what());
        return !str_value_.empty();
      }
    }
    case INTS: {
      return num_value_.int_value_ != 0;
    }
    case FLOATS: {
      float val = num_value_.float_value_;
      return val >= EPSILON || val <= -EPSILON;
    }
    case BOOLEANS: {
      return num_value_.bool_value_;
    }
    case DATES: {
      LOG_ERROR("can't convert date to boolean");
      return false;
    }
    default: {
      LOG_WARN("unknown data type. type=%d", attr_type_);
      return false;
    }
  }
  return false;
}

int Value::get_date() const {
  switch (attr_type_) {
    case CHARS:
    case INTS:
    case FLOATS:
    case BOOLEANS: {
      LOG_ERROR("can't convert other type to date");
      return 0;
    }
    case DATES: {
      return num_value_.date_value_;
    }
    default: {
      LOG_WARN("unknown data type. type=%d", attr_type_);
      return 0;
    }
  }
  return 0;
}

PageNum Value::get_page_num() const {
  if (attr_type_ == TEXTS) {
    return num_value_.page_num_value_;
  }
  return 0;
}

const char *Value::get_chars() const {
  return str_value_.c_str();
}
